package cn.itcast.util;

import cn.itcast.config.QuotConfig;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.lang.reflect.MalformedParameterizedTypeException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * 1.静态代码块获取连接对象
 * 2.获取表
 * 3.插入单列数据
 * 4.插入多列数据
 * 5.根据rowkey查询数据
 * 6.根据rowkey删除数据
 * 7.批量数据插入
 */
public class HbaseUtil {
    static Connection connection=null;

    //todo 1.静态代码块获取连接对象
    static {
        //设置zookeeper
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", QuotConfig.config.getProperty("zookeeper.connect"));
        //获取连接对象
        try {
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    //todo 2.获取表
    public static Table getTable(String tableName){
        Table table=null;
        try {
            table = connection.getTable(TableName.valueOf(tableName));
        } catch (IOException e) {
            e.printStackTrace();
        }
        return table;
    }
    //todo 3.插入单列数据
    public static void putDataByRowkey(String tableName,String rowkey,String family,String colName,String colValue){
        Table table = getTable(tableName);
        try {
            //封装数据
            Put put = new Put(Bytes.toBytes(rowkey));
            put.addColumn(Bytes.toBytes(family), Bytes.toBytes(colName), Bytes.toBytes(colValue));
            table.put(put);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                table.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    //todo 4.插入多列数据
    public static void putMapDataByRowkey(String tableName, String rowkey, String family, Map<String,Object> map) {
        Table table = getTable(tableName);

        try {
            Put put = new Put(rowkey.getBytes());
            for (String key : map.keySet()) {
                put.addColumn(Bytes.toBytes(family),key.getBytes(),map.get(key).toString().getBytes());
            }
            table.put(put);
        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            try {
                table.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    //todo 5.根据rowkey查询数据
    public static String queryByRowkey(String tableName,String rowkey,String family,String colName){
        Table table = getTable(tableName);
        String str=null;
        try {
            Get get = new Get(Bytes.toBytes(rowkey));
            Result result = table.get(get);
            byte[] value = result.getValue(family.getBytes(), colName.getBytes());
            str = new String(value);
        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            try {
                table.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return str;
    }
    //todo 6.根据rowkey删除数据
    public static void deleteByRowkey(String tableName,String rowkey,String family){
        Table table = getTable(tableName);
        try {
            Delete delete = new Delete(rowkey.getBytes());
            delete.addFamily(family.getBytes());
            table.delete(delete);
        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            try {
                table.close();

            } catch (IOException e) {
                e.printStackTrace();
            }
        }

    }
    //todo 7.批量数据插入
    public static void putList(String tableName, List<Put> list){
        Table table = getTable(tableName);
        try {
            table.put(list);
        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            try {
                table.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

}
